package studio.raptor.databus.client;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.cmdb.Config;
import studio.raptor.cmdb.ConfigService;
import studio.raptor.databus.ActiveMQConnectionFactory;
import studio.raptor.databus.ActiveMQSession;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Session;
import studio.raptor.databus.client.config.BrokerGroup;

/**
 * 消费者配置，包含消费者所属业务中心标志，消费的主题及消费者个性化配置。<br><br>
 * <p>
 * <p>
 * 消费者默认配置如下：<br>
 * 会话不开启事务；<br>
 * 消费者自动ACK消息模式；<br>
 * 开启批量ACK功能；<br>
 * 批量ACK超时时间30s；<br>
 * 消息重发次数不受限制；<br>
 * 每个消费者预取消息数量200；<br>
 * 两次重连之间最长的时间间隔为10s(从10毫秒开始指数增加到10s)<br>
 * failover后的尝试重连次数<br>
 * </P>
 */
public class DatabusConsumerBuilder {
    private static Logger log = LoggerFactory.getLogger(DatabusConsumerBuilder.class);

    /**
     * 会话是否为事务性，消费者默认为否
     */
    private boolean transactedSession = false;

    /**
     * 消息ACK模式，默认值(1)为自动确认
     */
    private int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
    /**
     * 开启批量ACK优化
     */
    private boolean optimizeAcknowledge = false;
    /**
     * ACK消息超时时间，默认30s--当不开启ACK优化时，此值无意义
     */
    private int optimizeAcknowledgeTimeOut = 30000;
    /**
     * 消息重发次数，默认5，表示直到消息被ACK，否则一直重发。0表示不重传,-1表示一直重发
     */
    private int maximumRedeliveries = 5;
    /**
     * 默认预取值设定为200
     */
    private int prefetchSize = 1;
    /**
     * 两次重连之间最长时间间隔，10s.
     */
    private int maxReconnectDelay = 10000;
    /**
     * failover机制中的最大重连次数。-1表示一直重连。
     */
    private int maxReconnectAttempts = -1;

    /**
     * 客户端启动时最大重连次数，默认为5
     */
    private int startupMaxReconnectAttempts = 5;
    /**
     * 通用其他配置项
     */
    private String otherURLConfig = "&";


    private List<BrokerGroup> brokerGroups = new ArrayList<BrokerGroup>();


    private DatabusConfig databusConfig;

    public DatabusConsumerBuilder(DatabusConfig databusConfig) {
        this.databusConfig = databusConfig;
        this.brokerGroups = databusConfig.getBrokerGroups();
    }

    /**
     * 构造携带选择器的消费者。
     * <p>
     * 方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
     * 表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
     * <br/>
     * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
     *
     * </p>
     * @param databusConfig 配置实例。
     * @param consumerSelector 消费者过滤器，参数遵循SQL92语法
     */


    /**
     * 创建死信队列消费者-默认情况下为手动ACK消息
     *
     * @param eventName 事件名称
     * @param eventKindId  事件种类ID
     * @return
     * @throws JMSException
     */
    public DatabusConsumer createDLQ(String eventName, Integer eventKindId) throws JMSException {
        return createDLQ(eventName,eventKindId,null);
    }

    /**
     * 创建死信队列消费者-默认情况下为手动ACK消息,携带消息选择器
     * <p>
     *  方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
     *  表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
     * <br/>
     * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
     * </p>
     * @param eventName 事件名称
     * @param eventKindId  事件种类ID
     * @param consumerSelector 消费者消息选择器，参数遵循SQL92语法
     * @return
     * @throws JMSException
     */
    public DatabusConsumer createDLQ(String eventName, Integer eventKindId,String consumerSelector) throws JMSException {
        return create(eventName,eventKindId,true,true,consumerSelector);
    }

    /**
     * 创建需要客户端手工确认的消费者，此消费者应该使用{@link DatabusConsumer#receiveUnACKMessage(long)} 方法获取消息
     * @param eventName 事件名称
     * @param eventKindId 事件种类ID
     * @return
     * @throws JMSException
     */
    public DatabusConsumer createClientACK(String eventName,Integer eventKindId) throws JMSException {
        return createClientACK(eventName,eventKindId,null);
    }

    /**
     * 创建需要客户端手工确认的消费者，此消费者应该使用{@link DatabusConsumer#receiveUnACKMessage(long)} 方法获取消息
     * <p>
     *  方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
     *  表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
     * <br/>
     * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
     * </p>
     * @param eventName 事件名称
     * @param eventKindId 事件种类ID
     * @param consumerSelector 消息过滤器
     * @return
     * @throws JMSException
     */
    public DatabusConsumer createClientACK(String eventName,Integer eventKindId,String consumerSelector) throws JMSException {
        return create(eventName,eventKindId,false,true,consumerSelector);
    }

    /**
     * 创建默认配置的消费者-自动ACK消息
     * @param eventName 事件名称
     * @param eventKindId 事件种类ID
     * @return 返回消费者
     * @throws JMSException
     */
    public DatabusConsumer create(String eventName,Integer eventKindId) throws JMSException {
        return create(eventName,eventKindId,null);
    }

    /**
     * 创建默认配置的消费者-自动ACK消息
     * <p>
     *  方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
     *  表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
     * <br/>
     * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
     * </p>
     * @param eventName 事件名称
     * @param eventKindId 事件种类ID
     * @param  consumerSelector 消息事件选择器
     * @return 返回消费者
     * @throws JMSException
     */
    public DatabusConsumer create(String eventName,Integer eventKindId,String consumerSelector) throws JMSException {
        return create(eventName,eventKindId,false,false,consumerSelector);
    }

    /**
     * 创建消费者
     * @param eventName 消费者消费的事件名称
     * @param eventKindId 消费者消费的事件种类ID
     * @param isDLQ 是否是消费死信队列
     * @param isClientACK 消费者是否是自动ACK
     * @return
     * @throws JMSException
     */
    private DatabusConsumer create(String eventName,Integer eventKindId,boolean isDLQ,boolean isClientACK,String consumerSelector) throws JMSException {
        if( null == eventName || "".equals(eventName) ){
            throw new JMSException("您所需消费的事件名称为空，请正确填写!");
        }if( null == eventKindId  ){
            throw new JMSException("您所需消费的事件种类为空，请正确填写!");
        }
        String topicSalt = TopicSalt.createSalt(eventName,eventKindId);
        DatabusConsumer databusConsumer = new DatabusConsumer();
        databusConsumer.setDatabusConfig(databusConfig);
        //校验当前消费者是否有权限消费指定事件
        if( ! databusConsumer.checkRecieveAuthorization(eventKindId,eventName) ){
            throw new JMSException("您无权订阅事件:["+eventName+"-"+eventKindId+"],若需订阅请先申请");
        }
        //初始化消费者参数
        try {
            initConfig(isClientACK);
        } catch (IOException e) {
            log.error(e.getMessage());
            throw new JMSException(e.getMessage());
        }

        Integer brokerGroupId = this.databusConfig.getBizPubSubEventInfo().getSub().get(eventName+"_"+eventKindId);

        String userName = "";
        String password = "";
        String brokerURL = DatabusConfig.LOAD_CONFIG_FAILURE;
        for(BrokerGroup brokerGroup:this.brokerGroups){
            if (brokerGroupId.intValue() == brokerGroup.getBrokerGroupId()) {
                userName = brokerGroup.getUserName();
                password = brokerGroup.getPassword();
                brokerURL = brokerGroup.getBrokerURL();
                break;
            }
        }
        if(DatabusConfig.LOAD_CONFIG_FAILURE.equals(brokerURL)){
            throw new JMSException("当前事件消息未配置发送目的地或配置了错误的目的地");
        }

        // 配置连接工厂类
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setUserName(userName);
        connectionFactory.setPassword(password);
        String customizedBrokerURL = this.transform2URLConfig();
        connectionFactory.setBrokerURL(brokerURL + "?" + customizedBrokerURL);

        // 配置消费者客户端中的连接、会话
        databusConsumer.setConnection(connectionFactory.createConnection());
        databusConsumer.getConnection().start();
        databusConsumer.setSession(databusConsumer.getConnection().createSession(this.isTransactedSession(), this.getAcknowledgeMode()));
        databusConsumer.setEventSeq(eventKindId);
        databusConsumer.setEventName(eventName);

        // 通过连接和消费者的配置创建真正的消费者
        String destinationURLConfig = this.transform2DestinationConfig();
        String groupQueueName = "";
        if (isDLQ){
            //若为死信队列
            groupQueueName = DatabusConfig.DLQ_PRE;
        }
        groupQueueName +=
            DatabusConfig.GROUP_QUEUE_NAME_PRE +
                databusConfig.getBizKey() + "." +
                DatabusConfig.VIRTUAL_TOPIC_PRE + eventName + "_" + topicSalt + "_" + eventKindId;
        Destination destination =
                databusConsumer.getSession().createQueue(groupQueueName + "?" + destinationURLConfig);
        if (!Objects.equals(null,consumerSelector) && !Objects.equals("",consumerSelector)){
            databusConsumer.setMessageConsumer(databusConsumer.getSession().createConsumer(destination,consumerSelector));
        }else{
            databusConsumer.setMessageConsumer(databusConsumer.getSession().createConsumer(destination));
        }


        return databusConsumer;
    }

    /**
     * 依据加载配置方式初始化消费者配置
     */
    private void initConfig(boolean isClientACK) throws IOException {
        if( isClientACK ){
            this.setAcknowledgeMode(ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);
        }else{
            this.setAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
        }
        if( databusConfig.getLoadType() == DatabusConfigLoadType.LOAD_TYPE_CMDB ){
            //从配置中心加载消费者配置
            Config config = ConfigService.getConfig(DatabusConfig.CLIENT_CONFIG_PRE + databusConfig.getBizKey());
            this.setTransactedSession(config.getBooleanProperty("consumer.transactedSession", this.isTransactedSession()));
            this.setOptimizeAcknowledge(config.getBooleanProperty("consumer.optimizeAcknowledge", this.isOptimizeAcknowledge()));
            this.setOptimizeAcknowledgeTimeOut(config.getIntProperty("consumer.optimizeAcknowledgeTimeOut", this.getOptimizeAcknowledgeTimeOut()));
            this.setMaximumRedeliveries(config.getIntProperty("consumer.maximumRedeliveries", this.getMaximumRedeliveries()));
            this.setPrefetchSize(config.getIntProperty("consumer.prefetchSize", this.getPrefetchSize()));
            this.setMaxReconnectDelay(config.getIntProperty("consumer.maxReconnectDelay", this.getMaxReconnectDelay()));
            this.setMaxReconnectAttempts(config.getIntProperty("consumer.maxReconnectAttempts", this.getMaxReconnectAttempts()));
            this.setStartupMaxReconnectAttempts(config.getIntProperty("consumer.startupMaxReconnectAttempts", this.getStartupMaxReconnectAttempts()));
            this.setOtherURLConfig(config.getProperty("consumer.otherURLConfig", this.getOtherURLConfig()));
        }else{
            InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream(this.databusConfig.CONFIG_LOCAL_DIR + "consumer.properties");
            Properties properties = new Properties();
            properties.load(in);

            this.setTransactedSession("true".equals(properties.getProperty("consumer.transactedSession")));
            this.setOptimizeAcknowledge("true".equals(properties.getProperty("consumer.optimizeAcknowledge")));
            this.setOptimizeAcknowledgeTimeOut(Integer.valueOf(properties.getProperty("consumer.optimizeAcknowledgeTimeOut")));
            this.setMaximumRedeliveries(Integer.valueOf(properties.getProperty("consumer.maximumRedeliveries")));
            this.setPrefetchSize(Integer.valueOf(properties.getProperty("consumer.prefetchSize")));
            this.setMaxReconnectDelay(Integer.valueOf(properties.getProperty("consumer.maxReconnectDelay")));
            this.setMaxReconnectAttempts(Integer.valueOf(properties.getProperty("consumer.maxReconnectAttempts")));
            this.setStartupMaxReconnectAttempts(Integer.valueOf(properties.getProperty("consumer.startupMaxReconnectAttempts")));
            this.setOtherURLConfig(properties.getProperty("consumer.otherURLConfig"));
        }

    }

    boolean isTransactedSession() {
        return transactedSession;
    }

    /**
     * 设定会话是否为事务性，消费者默认为否
     *
     * @param transactedSession
     */
    void setTransactedSession(boolean transactedSession) {
        this.transactedSession = transactedSession;
    }

    int getAcknowledgeMode() {
        return acknowledgeMode;
    }

    /**
     * 消息ACK模式
     *
     * @param acknowledgeMode 默认(1)为自动确认模式。2为手工批量ACK，4为手工单条消息确认。
     */
    void setAcknowledgeMode(int acknowledgeMode) {
        this.acknowledgeMode = acknowledgeMode;
    }

    boolean isOptimizeAcknowledge() {
        return optimizeAcknowledge;
    }

    /**
     * 设置是否开启批量ACK操作(当且仅当session不是事务性且ACK模式为自动确认时，开启此功能才有意义)
     *
     * @param optimizeAcknowledge 默认开启
     */
    void setOptimizeAcknowledge(boolean optimizeAcknowledge) {
        this.optimizeAcknowledge = optimizeAcknowledge;
    }

    int getOptimizeAcknowledgeTimeOut() {
        return optimizeAcknowledgeTimeOut;
    }

    /**
     * 设置批量ACK超时时间(当且仅当session不是事务性且ACK模式为自动确认时，此值才有意义)
     *
     * @param optimizeAcknowledgeTimeOut ACK超时时间，单位毫秒
     */
    void setOptimizeAcknowledgeTimeOut(int optimizeAcknowledgeTimeOut) {
        this.optimizeAcknowledgeTimeOut = optimizeAcknowledgeTimeOut;
    }

    int getMaximumRedeliveries() {
        return maximumRedeliveries;
    }

    /**
     * 设置消息最大重发次数(建议使用默认值)
     *
     * @param maximumRedeliveries 默认为-1，表示一直会重发
     */
    void setMaximumRedeliveries(int maximumRedeliveries) {
        this.maximumRedeliveries = maximumRedeliveries;
    }

    int getPrefetchSize() {
        return prefetchSize;
    }

    /**
     * 设置消费者的预取消息数量
     *
     * @param prefetchSize 默认200
     */
    void setPrefetchSize(int prefetchSize) {
        this.prefetchSize = prefetchSize;
    }

    int getMaxReconnectDelay() {
        return maxReconnectDelay;
    }

    /**
     * 设置最大重来延迟毫秒数
     *
     * @param maxReconnectDelay 默认10s
     */
    void setMaxReconnectDelay(int maxReconnectDelay) {
        this.maxReconnectDelay = maxReconnectDelay;
    }

    int getMaxReconnectAttempts() {
        return maxReconnectAttempts;
    }

    /**
     * 设置failover机制最大重连次数
     *
     * @param maxReconnectAttempts 默认值为-1，表示一直尝试重连。0表示不重连
     */
    void setMaxReconnectAttempts(int maxReconnectAttempts) {
        this.maxReconnectAttempts = maxReconnectAttempts;
    }

    String getOtherURLConfig() {
        return otherURLConfig;
    }

    /**
     * 设置其他参数，默认情况下不使用此项
     *
     * @param otherURLConfig
     */
    void setOtherURLConfig(String otherURLConfig) {
        this.otherURLConfig = otherURLConfig;
    }

    int getStartupMaxReconnectAttempts() {
        return startupMaxReconnectAttempts;
    }

    /**
     * 设置客户端启动时尝试连接服务器的最大次数，默认为5
     *
     * @param startupMaxReconnectAttempts 客户端启动时尝试连接服务器的最大次数
     */
    void setStartupMaxReconnectAttempts(int startupMaxReconnectAttempts) {
        this.startupMaxReconnectAttempts = startupMaxReconnectAttempts;
    }

    /**
     * 将消费者客户端配置转换为URL参数
     *
     * @return
     */
    public String transform2URLConfig() {
        StringBuilder urlConfig = new StringBuilder();
        urlConfig.append("jms.optimizeAcknowledge=" + this.isOptimizeAcknowledge());
        urlConfig.append("&jms.optimizeAcknowledgeTimeOut=" + this.getOptimizeAcknowledgeTimeOut());
        urlConfig.append("&jms.redeliveryPolicy.maximumRedeliveries=" + this.getMaximumRedeliveries());
        urlConfig.append("&maxReconnectDelay=" + this.getMaxReconnectDelay());
        urlConfig.append("&maxReconnectAttempts=" + this.getMaxReconnectAttempts());
        urlConfig.append("&startupMaxReconnectAttempts=" + this.getStartupMaxReconnectAttempts());
        urlConfig.append(this.getOtherURLConfig());
        return urlConfig.toString();
    }

    /**
     * 将消费者的消息目的地配置转换
     *
     * @return
     */
    public String transform2DestinationConfig() {
        StringBuilder destinationURLConfig = new StringBuilder();
        destinationURLConfig.append("consumer.prefetchSize=" + this.getPrefetchSize());
        return destinationURLConfig.toString();
    }


}
